package org.sqp.proxy;

import org.sqp.proxy.model.NetProxy;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author sunqipeng
 * @date 2019/1/2
 */
public abstract class AbstractNetProxyService implements NetProxyService {
    private LinkedBlockingQueue<NetProxy> proxyQueue = new LinkedBlockingQueue<>(1000);
    private Lock acqLock = new ReentrantLock(true);
    private volatile long lastAcq = 0;
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    class NetAcqRunner implements Runnable {
        public void run() {
            try {
                if (proxyQueue.size() == 0 || proxyQueue.size() < queueLength() * getLoadFactor()) {
                    long cur = System.currentTimeMillis();
                    long minIntervalNano = getMinInterval() * 1000;
                    if (cur - lastAcq < minIntervalNano) {
                        long slp = cur - lastAcq >= 0 ? (lastAcq + minIntervalNano - cur) / 1000 + 1 : 1;
                        Thread.sleep(slp);
                    }
                    System.out.println(Thread.currentThread().getName() + " acq at " + cur);
                    lastAcq = cur;
                    List<NetProxy> netProxies = netAcquire();
                    for (int i = netProxies.size() - 1; i > 0; i--) {
                        proxyQueue.offer(netProxies.get(i));
                    }
                    System.out.println(Thread.currentThread().getName() + " acq after left " + proxyQueue.size());
                }
            } catch (Exception e) {

            }
        }
    }

    private void checkQueue() {
        int size = proxyQueue.size();
        if (size == 0 || size < queueLength() * getLoadFactor()) {
            long cur = System.currentTimeMillis();
            long minInterval = getMinInterval();
            if (cur - lastAcq < minInterval){
                return;
            }
            try {
                acqLock.lock();
                if (proxyQueue.size() == 0 || proxyQueue.size() < queueLength() * getLoadFactor()) {
                    if (cur - lastAcq >= minInterval) {
                        lastAcq = cur;
                        executorService.execute(new NetAcqRunner());
                    }
                }
            } catch (Exception e){
                //
            }finally {
                acqLock.unlock();
            }
        }

    }

    public NetProxy blockingAcquire() {
        try {
            checkQueue();
            NetProxy p = proxyQueue.poll();
            if (p == null){
                Thread.sleep(100);
                return blockingAcquire();
            }
            return p;
        } catch (Exception e) {
            throw new RuntimeException("acquire proxy fail ", e);
        }
    }

    public NetProxy acquire() {
        return proxyQueue.peek();
    }

    public long queueLength() {
        return 100;
    }

    public float getLoadFactor() {
        return 0.75f;
    }

    //单位 毫秒
    public abstract long getMinInterval();

    public abstract List<NetProxy> netAcquire();
}
